package morefun.demo

import org.apache.spark.{SparkConf, SparkContext}

object ActionOperation {

  // 1. resuce 首先将第一个与第二个元素做 reduce 中定义的操作，然后把结果再与第三个元素做该操作，以此类推
  def reduce( sc:SparkContext): Unit = {

    val numList = sc.parallelize(Array(1,2,3,4,5,6,7,8,9,10))
    val sum = numList.reduce(_ + _);

    println("sum=" + sum)
  }

  // 2. collection 是将 RDD 数据从远程服务器获取到本地。一般不建议使用，因为要走网络传输。并且当数据量大的时候会导致内存溢出
  def collection(sc:SparkContext): Unit = {
    val numList = sc.parallelize(Array(1,2,3,4,5,6,7,8,9,10))
    val doubleNums = numList.map(num => num*2)

    // 使用 collection 把远程的 RDD 拉取到本地，变成本地数据
    val doubleNumArray = doubleNums.collect()
    // 再通过本地方法遍历数据，这里只是演示 collect ，正常还是应该用 foreach 来代替 collect
    for(num <- doubleNumArray) {
      println("num=" + num)
    }
  }

  // 3. count 获取 RDD 中元素个数
  def count(sc:SparkContext): Unit = {

    val numList = sc.parallelize(Array(1,2,3,4,5,6,7,8,9,10))
    val count = numList.count()
    println("count=" + count)
  }


  // 3. take 是将 RDD 数据从远程服务器获取前几条数据到本地。与collect类似，只是获取的是部分数据
  def take(sc:SparkContext): Unit = {
    val numList = sc.parallelize(Array(1,2,3,4,5,6,7,8,9,10))
    val doubleNums = numList.map(num => num*2)

    // 使用 take 把远程的 RDD 中的前3条，拉取到本地，变成本地数据
    val doubleNumArray = doubleNums.take(3)
    // 再通过本地方法遍历数据
    for(num <- doubleNumArray) {
      println("num=" + num)
    }
  }

  // 4. saveAsTextfile 可以将 RDD 保存到 hdfs 或者本地文件
  // 保存的是个文件夹形式，里面有个 part-00000 是最后真是的数据
  def saveAsTextfile(sc:SparkContext): Unit = {
    val numList = sc.parallelize(Array(1,2,3,4,5,6,7,8,9,10))
    val doubleNums = numList.map(num => num*2)

    // 使用 collection 把远程的 RDD 拉取到本地，变成本地数据
    val doubleNumArray = doubleNums.saveAsTextFile("file:/Users/wdy/tmp/test")
//    val doubleNumArray = doubleNums.saveAsTextFile("hdfs://spark1:9000/doublenums")

  }

  // 5. countByKey 统计每个key对应的元素个数，返回的是一个 map<string, object>，
  def countByKey(sc:SparkContext): Unit = {
    val scoresList = Array(Tuple2("class1", "name1"),Tuple2("class2","name1"),Tuple2("class3","name1"),Tuple2("class2","name2"),Tuple2("class1","name1"),Tuple2("class4","name1"))

    val scoresArray = sc.parallelize(scoresList)

    val classStudents = scoresArray.countByKey()
    println(classStudents)
    classStudents.foreach(count => {
      print("classname:" + count._1)
      print("count:" + count._2)
      print("=============================")
    })

    println(classStudents)
  }

  // 6. foreach 遍历 RDD，遍历过程是在远程完成的，不是在本地执行的，所以效率要比 collect 到本地再遍历效率高很多
  def foreach(sc:SparkContext): Unit = {
    val numList = Array(1,2,3,4,5,6)
    val numArray = sc.parallelize(numList)

    numArray.foreach( num => {
      print(num)
      println("-----------------")
    })

  }

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("ActionOperation ").setMaster("local")
    val sc = new SparkContext(conf)

//    reduce(s on(sc)
//    collection(sc)
//    count(sc)
//    take(sc)
//    saveAsTextfile(sc)
//    countByKey(sc)
    foreach(sc)
  }
}
